package org.lql.chaos.rocketmq.service.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;

/**
 * @author: liangqinglong
 * @date: 2025-08-17 00:40
 * @description: 通用消息消费者，处理所有类型的消息
 **/
@Slf4j
@Service
@RocketMQMessageListener(
        topic = "test-topic",
        consumerGroup = "universal-consumer-group"
)
public class UniversalConsumer implements RocketMQListener<MessageExt> {

    @Override
    public void onMessage(MessageExt message) {
        String msg = new String(message.getBody(), StandardCharsets.UTF_8);
        String tag = message.getTags();
        log.info("通用消费者接收到消息: {}，标签: {}，消息ID: {}", msg, tag, message.getMsgId());
        
        // 根据标签处理不同类型的消息
        if ("normal".equals(tag)) {
            handleNormalMessage(msg);
        } else if ("order".equals(tag)) {
            handleOrderMessage(msg);
        } else if ("delay".equals(tag)) {
            handleDelayMessage(msg);
        } else if ("transaction".equals(tag)) {
            handleTransactionMessage(msg);
        } else if ("batch".equals(tag)) {
            handleBatchMessage(msg);
        } else if ("oneway".equals(tag)) {
            handleOneWayMessage(msg);
        } else if ("async".equals(tag)) {
            handleAsyncMessage(msg);
        } else {
            handleUnknownMessage(msg, tag);
        }
    }
    
    private void handleNormalMessage(String message) {
        log.info("处理普通消息: {}", message);
    }
    
    private void handleOrderMessage(String message) {
        log.info("处理顺序消息: {}", message);
    }
    
    private void handleDelayMessage(String message) {
        log.info("处理延时消息: {}，处理时间: {}", message, System.currentTimeMillis());
    }
    
    private void handleTransactionMessage(String message) {
        log.info("处理事务消息: {}", message);
    }
    
    private void handleBatchMessage(String message) {
        log.info("处理批量消息中的一条: {}", message);
    }
    
    private void handleOneWayMessage(String message) {
        log.info("处理单向消息: {}", message);
    }
    
    private void handleAsyncMessage(String message) {
        log.info("处理异步消息: {}", message);
    }
    
    private void handleUnknownMessage(String message, String tag) {
        log.warn("处理未知类型消息: {}，标签: {}", message, tag);
    }
}